/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hexiang.rocketmq.producer.transaction.listener;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hexiang.rocketmq.message.DemoMessage;
import com.hexiang.rocketmq.service.DemoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class DemoTransactionListenerImpl implements TransactionListener {

    @Resource
    private DemoService demoService;
    @Resource
    private ObjectMapper objectMapper;

    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
        LocalTransactionState localTransactionState = LocalTransactionState.COMMIT_MESSAGE;
        //获取消息体
        String jsonString = new String(message.getBody());
        log.info("demoExecuteLocalTransactionStart,message is {}", jsonString);
        if (StringUtils.isNotBlank(jsonString)) {
            DemoMessage demoMessage;
            try {
                demoMessage = objectMapper.readValue(jsonString, DemoMessage.class);
            } catch (JsonProcessingException e) {
                throw new IllegalArgumentException("事务消息消息转对象失败");
            }
            //判断是否为空
            if (null == demoMessage) {
                throw new IllegalArgumentException("事务消息消息对象不能为空");
            }
            try {
                //保存，事务方法
                demoService.saveDemo(demoMessage);
            } catch (RuntimeException e) {
                log.error("executeLocalTransactionRuntimeException,message is{}", message, e);
                localTransactionState = LocalTransactionState.ROLLBACK_MESSAGE;
            } catch (Exception e) {
                log.error("executeLocalTransactionException,message is{}", message, e);
                localTransactionState = LocalTransactionState.UNKNOW;
            }
        } else {
            throw new IllegalArgumentException("事务消息消息对象不能为空");
        }

        log.info("demoExecuteLocalTransactionEnd,message is {},transactionState is {}", message, localTransactionState);

        return localTransactionState;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt message) {
        String jsonString = new String(message.getBody());

        log.info("demoCheckLocalTransactionStart,message is {}", jsonString);

        LocalTransactionState transactionState = LocalTransactionState.ROLLBACK_MESSAGE;

        if (StringUtils.isNotBlank(jsonString)) {
            DemoMessage demoMessage;
            //获取消息对象
            try {
                demoMessage = objectMapper.readValue(jsonString, DemoMessage.class);
            } catch (JsonProcessingException e) {
                throw new IllegalArgumentException("事务消息消息转对象失败");
            }

            Long id;
            if (null != demoMessage) {
                //入库单号
                id = demoMessage.getId();
                if (null != id) {
                    try {
                        //检查消息是否存在
                        if (demoService.checkDemoMessageExist(demoMessage)) {
                            transactionState = LocalTransactionState.COMMIT_MESSAGE;
                        }
                    } catch (Exception e) {
                        log.error("checkLocalTransactionException", e);
                        transactionState = LocalTransactionState.UNKNOW;
                    }
                } else {
                    log.warn("checkLocalTransactionDemoMessageIsNull,messageObj is {}", demoMessage);
                }
            } else {
                log.warn("checkLocalTransactionDemoMessageIsNoValid,message is {}", jsonString);
            }
        } else {
            log.warn("checkLocalTransactionDemoMessageIsNull,message is {}", jsonString);
        }

        log.info("demoCheckLocalTransactionEnd,message is {},transactionState is {}", message, transactionState);

        return transactionState;
    }
}
